package cn.openea.perpartner.meily.rabbit.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.openea.perpartner.common.core.manager.AsyncManager;
import cn.openea.perpartner.common.core.text.Convert;
import cn.openea.perpartner.meily.rabbit.domain.BusinessDto;
import cn.openea.perpartner.meily.rabbit.domain.DelayDto;
import cn.openea.perpartner.meily.rabbit.factory.BusinessServiceFactory;
import cn.openea.perpartner.meily.rabbit.factory.DelayServiceFactory;
import cn.openea.perpartner.meily.rabbit.service.BusinessService;
import cn.openea.perpartner.meily.rabbit.service.DelayService;
import cn.openea.perpartner.meily.web.entity.Goods;
import cn.openea.perpartner.meily.web.service.IGoodsService;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author Weird077
 * @description TODO
 * @date 2023-02-07 16:35
 */
@Slf4j
@Component
public class ConsumerService {
    @Autowired
    private IGoodsService goodsService;
    @Autowired
    private PublisherService publisher;


    /**
     * 消费延时队列
     *
     * @param object
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "${rabbit.delay.queue}")
    public void receiveDelay(JSONObject object, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        DelayDto delayDto = JSON.toJavaObject(object, DelayDto.class);
        try {
            Goods goods = goodsService.getById(delayDto.getGoodsId());
            //1、判断商品是否存在，商品是否已上架，若已下架则不再执行业务逻辑，也不再重新入队
            if (BeanUtil.isNotEmpty(goods) && "已上架".equals(goods.getStatus())) {
                //2、计算新的周期串
                String cyclicId = goodsService.cycle(goods.getGoodsId());
                //3、判断队列里的周期串和重新计算的周期串是否属于同一个周期，若不是则不再执行业务逻辑，也不再重新入队
                if (goodsService.isSameCycle(delayDto.getCyclicId(), cyclicId)) {
                    //4、通过beanName获取对应的延时处理类
                    DelayService delayService = DelayServiceFactory.getHandler(delayDto.getBeanName());
                    //5、执行延时处理类的业务逻辑
                    AsyncManager.me().execute(delayService.execute(delayDto));
                    //6、取周期天数
                    String cycle = goods.getHelpCycle();
                    //7、取发布时间
                    DateTime releaseTime = DateUtil.parseDateTime(goods.getReleaseTime());
                    //8、重新计算周期到期时间
                    DateTime nextTime = DateUtil.offsetDay(releaseTime, Convert.toInt(cycle));
                    //9、计算队列到期时间，精确到毫秒
                    long delayTime = DateUtil.betweenMs(releaseTime, nextTime);
                    delayDto.setCyclicId(cyclicId);
                    delayDto.setDelayTime(delayTime);
                    //10、重新入队
                    publisher.execute(delayDto);
                }
            }
            //11、消息消费成功，手动确认消息,消息从队列中删除
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            //消息消费失败，手动拒绝消息,消息从队列中删除
            channel.basicNack(deliveryTag, false, false);
            log.error(e.getMessage());
        }
    }

    /**
     * 消费业务队列
     *
     * @param object
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "${rabbit.business.queue}")
    public void receiveBusiness(JSONObject object, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        BusinessDto businessDto = JSON.toJavaObject(object, BusinessDto.class);
        try {
            //1、通过beanName获取对应的延时处理类
            BusinessService businessService = BusinessServiceFactory.getHandler(businessDto.getBeanName());
            //2、执行延时处理类的业务逻辑
            AsyncManager.me().execute(businessService.execute(businessDto));
            //3、消息消费成功，手动确认消息,消息从队列中删除
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error(e.getMessage());
            //消息消费失败，手动拒绝消息,消息从队列中删除
            channel.basicNack(deliveryTag, false, false);
        }
    }
}
